// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: yuxuan

#pragma once

#include <algorithm>
#include <cmath>
#include <cstdint>
#include <deque>
#include <memory>
#include <unordered_map>
#include <utility>

#include "coin2/base/log.h"

namespace coin2::strategy::util {

constexpr int64_t DEFAULT_MEM_EFFICIENT_THRESHOLD = 14400'000'000'000LL + 1LL;

template <typename T, typename V>
class MovingWindow {
 public:
  explicit MovingWindow(const int64_t& timewindow) : timewindow_(timewindow) {
    if (timewindow_ == 0) {
      timewindow_ = 1LL;
    }
  }
  virtual ~MovingWindow() = default;

  int Size() const { return dq_.size(); }
  bool Empty() const { return Size() == 0; }
  virtual void Add(int64_t ts, const T& obj) = 0;
  virtual void Add(int64_t ts, T&& obj) { Add(ts, std::forward<const T>(obj)); }
  virtual void Update(int64_t ts) = 0;
  const V& GetLastData() const {
    CHECK(!dq_.empty());
    return dq_.back().second;
  }
  const V& GetFirstData() const {
    CHECK(!dq_.empty());
    return dq_.front().second;
  }
  int64_t GetLastTs() const {
    CHECK(!dq_.empty());
    return dq_.back().first;
  }
  int64_t GetFirstTs() const {
    CHECK(!dq_.empty());
    return dq_.front().first;
  }
  int64_t GetWindowSize() const { return timewindow_; }
  bool Ready() const { return ready_; }

  void ReplaceLast(int64_t ts, const T& obj) {
    if (std::isfinite(static_cast<double>(obj))) {
      PopLast();
      Add(ts, obj);
    }
  }

 protected:
  virtual void PopLast() { CHECK(false); }

  int64_t timewindow_;
  std::deque<std::pair<int64_t, V>> dq_;
  bool ready_{false};
};

template <typename T>
class MovingSum : public MovingWindow<T, T> {
 public:
  explicit MovingSum(const int64_t& timewindow, int64_t multiplier = 1)
      : MovingWindow<T, T>(timewindow) {}

  virtual void Add(int64_t ts, const T& obj) {
    Update(ts);
    if (std::isfinite(static_cast<double>(obj))) {
      sum_ += obj;
      MovingWindow<T, T>::dq_.push_back({ts, obj});
    }
  }

  virtual void Update(int64_t ts) {
    while (!MovingWindow<T, T>::Empty() &&
           ts - MovingWindow<T, T>::dq_.front().first > MovingWindow<T, T>::timewindow_) {
      sum_ -= MovingWindow<T, T>::dq_.front().second;
      MovingWindow<T, T>::dq_.pop_front();
      MovingWindow<T, T>::ready_ = true;
    }
    if (MovingWindow<T, T>::dq_.empty()) {
      // prevent floating error when all values are poped
      sum_ = 0;
    }
  }

  virtual T GetSum() { return T(sum_); }

 protected:
  void PopLast() override {
    if (!MovingWindow<T, T>::Empty()) {
      sum_ -= MovingWindow<T, T>::dq_.back().second;
      MovingWindow<T, T>::dq_.pop_back();
    }
    if (MovingWindow<T, T>::dq_.empty()) {
      // prevent floating error when all values are poped
      sum_ = 0;
    }
  }

 private:
  T sum_{0};
};

template <typename T>
class MemMovingSum : public MovingSum<T> {
 public:
  explicit MemMovingSum(const int64_t& timewindow, int64_t merge_window)
      : MovingSum<T>(timewindow), merge_window_(merge_window) {
    CHECK_GT(merge_window_, 0);
  }

  void Add(int64_t ts, const T& obj) override {
    Update(ts);
    if (std::isfinite(static_cast<double>(obj))) {
      sum_ += obj;
      int64_t interval = ts / merge_window_;
      CHECK_GE(interval, interval_id_);
      if (interval == interval_id_) {
        MovingWindow<T, T>::dq_.back().second += obj;
      } else {
        interval_id_ = interval;
        MovingWindow<T, T>::dq_.push_back({ts, obj});
      }
    }
  }

  void Update(int64_t ts) override {
    real_last_ts_ = ts;
    while (!MovingWindow<T, T>::Empty() && ts - MovingWindow<T, T>::dq_.front().first >
                                               MovingWindow<T, T>::timewindow_ + merge_window_) {
      sum_ -= MovingWindow<T, T>::dq_.front().second;
      MovingWindow<T, T>::dq_.pop_front();
      MovingWindow<T, T>::ready_ = true;
    }
    if (MovingWindow<T, T>::dq_.empty()) {
      // prevent floating error when all values are poped
      sum_ = 0;
    }
  }

  T GetSum() override {
    if (!MovingWindow<T, T>::Empty()) {
      int64_t exceed_ts =
          real_last_ts_ - MovingWindow<T, T>::GetFirstTs() - MovingWindow<T, T>::timewindow_;
      CHECK_LE(exceed_ts, merge_window_);
      if (exceed_ts <= 0) {
        return sum_;
      } else {
        double ratio = static_cast<double>(exceed_ts) / static_cast<double>(merge_window_);
        double res = static_cast<double>(sum_) -
                     ratio * static_cast<double>(MovingWindow<T, T>::GetFirstData());
        return static_cast<T>(res);
      }
    }
    return sum_;
  }

 protected:
  void PopLast() override {
    CHECK(false);  // Disable for now
  }

 private:
  T sum_{0};
  int64_t merge_window_;
  int64_t real_last_ts_{0};
  int64_t interval_id_{0};
};

template <typename T>
class MovingAverage : public MovingWindow<T, T> {
 public:
  explicit MovingAverage(const int64_t& timewindow, int64_t multiplier = 1)
      : MovingWindow<T, T>(timewindow) {}

  virtual void Add(int64_t ts, const T& obj) {
    Update(ts);
    if (std::isfinite(static_cast<double>(obj))) {
      sum_ += obj;
      T obj22 = obj * obj;
      sum2_ += obj22;
      MovingWindow<T, T>::dq_.push_back({ts, obj});
      dq_sqr_.push_back(obj22);
    }
  }

  virtual void Update(int64_t ts) {
    while (!MovingWindow<T, T>::Empty() &&
           ts - MovingWindow<T, T>::dq_.front().first > MovingWindow<T, T>::timewindow_) {
      sum_ -= MovingWindow<T, T>::dq_.front().second;
      sum2_ -= dq_sqr_.front();
      MovingWindow<T, T>::dq_.pop_front();
      dq_sqr_.pop_front();
      MovingWindow<T, T>::ready_ = true;
    }
  }

  virtual bool GetStatistics(double* mean, double* std) {
    int n = MovingWindow<T, T>::Size();
    if (n > 0) {
      double m = static_cast<double>(sum_) / n;
      double s = static_cast<double>(sum2_) / n - m * m;
      *mean = m;
      *std = std::sqrt(std::max(s, 0.0));
      return true;
    }
    sum_ = sum2_ = *mean = *std = 0;
    return false;
  }

 protected:
  void PopLast() override {
    if (!MovingWindow<T, T>::Empty()) {
      sum_ -= MovingWindow<T, T>::dq_.back().second;
      sum2_ -= dq_sqr_.back();
      MovingWindow<T, T>::dq_.pop_back();
      dq_sqr_.pop_back();
    }
  }

 private:
  T sum_{0};
  T sum2_{0};
  std::deque<T> dq_sqr_;
};

template <typename T>
class MemMovingAverage : public MovingAverage<T> {
 public:
  explicit MemMovingAverage(const int64_t& timewindow, int64_t merge_window)
      : MovingAverage<T>(timewindow), merge_window_(merge_window) {
    CHECK_GT(merge_window_, 0);
  }

  void Add(int64_t ts, const T& obj) override {
    Update(ts);
    if (std::isfinite(static_cast<double>(obj))) {
      sum_ += obj;
      T obj22 = obj * obj;
      sum2_ += obj22;
      num_obj_ += 1;
      int64_t interval = ts / merge_window_;
      CHECK_GE(interval, interval_id_);
      if (interval == interval_id_) {
        MovingWindow<T, T>::dq_.back().second += obj;
        dq_sqr_.back() += obj22;
        bucket_counter_.back() += 1;
      } else {
        interval_id_ = interval;
        MovingWindow<T, T>::dq_.push_back({ts, obj});
        dq_sqr_.push_back(obj22);
        bucket_counter_.push_back(1);
      }
    }
  }

  void Update(int64_t ts) override {
    while (!MovingWindow<T, T>::Empty() && ts - MovingWindow<T, T>::dq_.front().first >
                                               (MovingWindow<T, T>::timewindow_ + merge_window_)) {
      sum_ -= MovingWindow<T, T>::dq_.front().second;
      sum2_ -= dq_sqr_.front();
      MovingWindow<T, T>::dq_.pop_front();
      dq_sqr_.pop_front();
      num_obj_ -= bucket_counter_.front();
      bucket_counter_.pop_front();
      MovingWindow<T, T>::ready_ = true;
    }
  }

  bool GetStatistics(double* mean, double* std) override {
    int n = num_obj_;
    if (n > 0) {
      double m = static_cast<double>(sum_) / n;
      double s = static_cast<double>(sum2_) / n - m * m;
      *mean = m;
      *std = std::sqrt(std::max(s, 0.0));
      return true;
    }
    sum_ = sum2_ = *mean = *std = 0;
    return false;
  }

  void PopLast() override {
    CHECK(false);  // Disable for now
  }

 private:
  T sum_{0};
  T sum2_{0};
  std::deque<T> dq_sqr_;
  std::deque<int> bucket_counter_;
  int64_t merge_window_;
  int64_t interval_id_{0};
  int num_obj_{0};
};

template <typename T>
class MovingUnique : public MovingWindow<T, int64_t> {
 public:
  explicit MovingUnique(const int64_t& timewindow) : MovingWindow<T, int64_t>(timewindow) {}

  void Add(int64_t ts, const T& obj) final {
    Update(ts);
    item_cnt_[obj]++;
    MovingWindow<T, int64_t>::dq_.emplace_back(ts, obj);
  }

  void Update(int64_t ts) final {
    while (!MovingWindow<T, int64_t>::Empty() && ts - MovingWindow<T, int64_t>::dq_.front().first >
                                                     MovingWindow<T, int64_t>::timewindow_) {
      const T& key = MovingWindow<T, int64_t>::dq_.front().second;
      item_cnt_[key]--;
      if (item_cnt_[key] == 0) {
        item_cnt_.erase(key);
      }
      MovingWindow<T, int64_t>::dq_.pop_front();
    }
  }

  size_t GetUnique() { return item_cnt_.size(); }

 private:
  std::unordered_map<T, int> item_cnt_;
};

template <typename T>
class MovingObject : public MovingWindow<T, T> {
 public:
  explicit MovingObject(const int64_t& timewindow) : MovingWindow<T, T>(timewindow) {}

  void Add(int64_t ts, const T& obj) final {
    Update(ts);
    MovingWindow<T, T>::dq_.push_back({ts, obj});
  }

  void PopLast() override {
    if (!MovingWindow<T, T>::Empty()) {
      MovingWindow<T, T>::dq_.pop_back();
    }
  }

  void Update(int64_t ts) final {
    while (!MovingWindow<T, T>::Empty() &&
           ts - MovingWindow<T, T>::dq_.front().first > MovingWindow<T, T>::timewindow_) {
      MovingWindow<T, T>::dq_.pop_front();
      MovingWindow<T, T>::ready_ = true;
    }
  }
};

template <typename T>
class VolumeWeightedMovingAverage : public MovingWindow<T, T> {
 public:
  explicit VolumeWeightedMovingAverage(const int64_t& timewindow, int64_t multiplier = 1)
      : MovingWindow<T, T>(timewindow) {}

  void Add(int64_t ts, const T& obj) override { CHECK(false); }
  void Add(int64_t ts, T&& obj) override { CHECK(false); }

  virtual void Add(int64_t ts, const T& p, const T& q) {
    Update(ts);
    if (std::isfinite(static_cast<double>(p)) && std::isfinite(static_cast<double>(q))) {
      T pq = p * q;
      pq_ += pq;
      MovingWindow<T, T>::dq_.push_back({ts, pq});
      q_ += q;
      dq_q_.push_back(q);
    }
  }

  virtual void Update(int64_t ts) {
    while (!MovingWindow<T, T>::Empty() &&
           ts - MovingWindow<T, T>::dq_.front().first > MovingWindow<T, T>::timewindow_) {
      auto pq = MovingWindow<T, T>::dq_.front().second;
      pq_ -= pq;
      auto qq = dq_q_.front();
      q_ -= qq;
      MovingWindow<T, T>::dq_.pop_front();
      dq_q_.pop_front();
      MovingWindow<T, T>::ready_ = true;
    }
  }

  virtual double GetAverage() const { return 1.0 * pq_ / q_; }

 private:
  T pq_{0};
  T q_{0};
  std::deque<T> dq_q_;
};

template <typename T>
class MemVolumeWeightedMovingAverage : public VolumeWeightedMovingAverage<T> {
 public:
  explicit MemVolumeWeightedMovingAverage(const int64_t& timewindow, int64_t merge_window)
      : VolumeWeightedMovingAverage<T>(timewindow), merge_window_(merge_window) {
    CHECK_GT(merge_window_, 0);
  }

  void Add(int64_t ts, const T& p, const T& q) override {
    Update(ts);
    if (std::isfinite(static_cast<double>(p)) && std::isfinite(static_cast<double>(q))) {
      T pq = p * q;
      pq_ += pq;
      q_ += q;
      int64_t interval = ts / merge_window_;
      CHECK_GE(interval, interval_id_);
      if (interval == interval_id_) {
        MovingWindow<T, T>::dq_.back().second += pq;
        dq_q_.back() += q;
      } else {
        interval_id_ = interval;
        MovingWindow<T, T>::dq_.push_back({ts, pq});
        dq_q_.push_back(q);
      }
    }
  }

  void Update(int64_t ts) override {
    real_last_ts_ = ts;
    while (!MovingWindow<T, T>::Empty() && ts - MovingWindow<T, T>::dq_.front().first >
                                               MovingWindow<T, T>::timewindow_ + merge_window_) {
      auto pq = MovingWindow<T, T>::dq_.front().second;
      pq_ -= pq;
      auto qq = dq_q_.front();
      q_ -= qq;
      MovingWindow<T, T>::dq_.pop_front();
      dq_q_.pop_front();
      MovingWindow<T, T>::ready_ = true;
    }
  }

  double GetAverage() const override { return 1.0 * pq_ / q_; }

 private:
  T pq_{0};
  T q_{0};
  std::deque<T> dq_q_;
  int64_t merge_window_;
  int64_t real_last_ts_{0};
  int64_t interval_id_{0};
};

template <typename T>
class TimeWeightedMovingAverage : public MovingWindow<T, T> {
 public:
  explicit TimeWeightedMovingAverage(const int64_t& timewindow) : MovingWindow<T, T>(timewindow) {}

  void Update(int64_t ts) override { CHECK(false); }

  void Add(int64_t ts, const T& obj) override {
    while (!MovingWindow<T, T>::Empty()) {
      int64_t first_ts = MovingWindow<T, T>::GetFirstTs();
      if (ts - first_ts <= MovingWindow<T, T>::timewindow_) {
        break;
      }
      T first_data = MovingWindow<T, T>::GetFirstData();
      MovingWindow<T, T>::dq_.pop_front();
      MovingWindow<T, T>::ready_ = true;
      if (!MovingWindow<T, T>::Empty()) {
        int64_t second_ts = MovingWindow<T, T>::GetFirstTs();
        tv_ -= (second_ts - first_ts) * first_data;
      }
    }
    if (std::isfinite(static_cast<double>(obj))) {
      if (!MovingWindow<T, T>::Empty()) {
        tv_ += static_cast<T>(ts - MovingWindow<T, T>::GetLastTs()) *
               MovingWindow<T, T>::GetLastData();
      }
      MovingWindow<T, T>::dq_.push_back({ts, obj});
    }
  }

  double GetAverage() const {
    if (!MovingWindow<T, T>::Empty()) {
      int64_t interval = MovingWindow<T, T>::GetLastTs() - MovingWindow<T, T>::GetFirstTs();
      if (interval > 0) {
        return static_cast<double>(tv_) / static_cast<double>(interval);
      } else {
        return MovingWindow<T, T>::GetLastData();
      }
    }
    return NAN;
  }

  double GetAverage(int64_t ts) const {
    if (!MovingWindow<T, T>::Empty()) {
      CHECK_GE(ts, (MovingWindow<T, T>::GetLastTs()));
      int64_t interval = ts - MovingWindow<T, T>::GetFirstTs();
      if (interval > 0) {
        T tv = tv_ + static_cast<T>(ts - MovingWindow<T, T>::GetLastTs()) *
                         MovingWindow<T, T>::GetLastData();
        return static_cast<double>(tv) / static_cast<double>(interval);
      } else {
        return MovingWindow<T, T>::GetLastData();
      }
    }
    return NAN;
  }

 private:
  T tv_{0};
};

template <typename T>
class FixedWindowExponentialMovingAverage {
 public:
  explicit FixedWindowExponentialMovingAverage(const int64_t& timewindow, double alpha)
      : alpha_(alpha) {
    mo_ = std::make_unique<MovingObject<double>>(timewindow);
  }

  void Add(int64_t ts, const T& obj) {
    Update(ts);
    if (std::isfinite(static_cast<double>(obj))) {
      if (!mo_->Empty()) {
        ema_ = ema_ * (1 - alpha_) + obj * alpha_;
      } else {
        ema_ = static_cast<double>(obj);
      }
      mo_->Add(ts, ema_);
    }
  }

  void Update(int64_t ts) { mo_->Update(ts); }

  double GetEma() const {
    double prev_ema_ = 0;
    if (mo_->Ready()) {
      prev_ema_ = mo_->GetFirstData();
    }
    int step = mo_->Size() - 1;
    double factor = std::pow(1 - alpha_, step);
    double ema = (ema_ - prev_ema_ * factor) / (1 - factor);
    return ema;
  }

  bool Ready() const { return mo_->Ready(); }

 private:
  double ema_{0};
  double alpha_;

  std::unique_ptr<MovingObject<double>> mo_;
};

template <typename T>
std::unique_ptr<MovingSum<T>> DispatchMovingSumInstance(
    int64_t time_window,
    int64_t merge_time_window) {
  if (time_window >= DEFAULT_MEM_EFFICIENT_THRESHOLD) {
    return std::move(std::make_unique<MemMovingSum<T>>(time_window, merge_time_window));
  } else {
    return std::move(std::make_unique<MovingSum<T>>(time_window));
  }
}

template <typename T>
std::unique_ptr<MovingAverage<T>> DispatchMovingAverageInstance(
    int64_t time_window,
    int64_t merge_time_window) {
  if (time_window >= DEFAULT_MEM_EFFICIENT_THRESHOLD) {
    return std::move(std::make_unique<MemMovingAverage<T>>(time_window, merge_time_window));
  } else {
    return std::move(std::make_unique<MovingAverage<T>>(time_window));
  }
}

template <typename T>
std::unique_ptr<VolumeWeightedMovingAverage<T>> DispatchMovingVwapInstance(
    int64_t time_window,
    int64_t merge_time_window) {
  if (time_window >= DEFAULT_MEM_EFFICIENT_THRESHOLD) {
    return std::move(
        std::make_unique<MemVolumeWeightedMovingAverage<T>>(time_window, merge_time_window));
  } else {
    return std::move(std::make_unique<VolumeWeightedMovingAverage<T>>(time_window));
  }
}

class ExponentialMovingAverage {
 public:
  ExponentialMovingAverage(double alpha, int64_t update_interval = 0)
      : alpha_(alpha), update_interval_(update_interval) {
    CHECK_GT(alpha, 0);
    CHECK_LT(alpha, 1);
  }

  void Update(int64_t ts, double v) {
    if (ts >= last_update_ts_ + update_interval_ && std::isfinite(v)) {
      if (init_) {
        ma_ = v * alpha_ + ma_ * (1 - alpha_);
      } else {
        ma_ = v;
        init_ = true;
      }
      last_update_ts_ = ts;
    }
  }

  double Get() const { return ma_; }

 private:
  double alpha_;
  int64_t update_interval_;
  int64_t last_update_ts_{0};
  double ma_;
  bool init_{false};
};

}  // namespace coin2::strategy::util
